package com.test.kfk.code;

import com.alibaba.fastjson.JSON;
import com.test.kfk.code.consume.KafkaConsumer;
import com.test.kfk.code.consume.User;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.*;


/**
 * @author chen_lei
 * @date 2020/11/16 20:10
 */
@RestController
public class SendController {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    public static final String topic8 = "topic8";

    @PostMapping("/send1")
    public void send1(@RequestBody User user)throws Exception{
        for (int i = 0; i < 20; i++) {
            SendResult<String, Object> result =
                    kafkaTemplate.send(KafkaConsumer.usertopic, String.valueOf(i), JSON.toJSONString(user)).get();

        }

        //指定分区
        kafkaTemplate.send(KafkaConsumer.usertopic,0,"key","val");
    }


    @GetMapping("/send8")
    public void send8(String msg){

        for (int i = 0; i < 5; i++) {
            kafkaTemplate.send(topic8,String.valueOf(new Random().nextInt(10)),String.valueOf(i));
        }

//        kafkaTemplate.send(topic8, msg);
//        System.out.println("send sync: " + msg);
//
//
//        kafkaTemplate.send(topic8, msg).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
//            @Override
//            public void onFailure(Throwable ex) {
//                System.out.println("发送消息失败："+ex.getMessage());
//            }
//
//            @Override
//            public void onSuccess(SendResult<String, Object> result) {
//                System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
//                        + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
//            }
//        });
//        System.out.println("send async: " + msg);
    }

    @GetMapping("transaction")
    public void transaction(){

        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(topic8,"test executeInTransaction");
            operations.send(topic8, "test executeInTransaction2");
            throw new RuntimeException("fail");
          //  return null;
        });

    }



//    @GetMapping("listTopic")
//    public void ss(){
//        Map<String, Object> propsMap = new HashMap<>();
//        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.195:9092");
//        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
//        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
//        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
//        propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
//
//        KafkaConsumer kafkaConsumer = new KafkaConsumer<String,String>(propsMap);
//        kafkaConsumer.subscribe(Arrays.asList("topic8"));
//        Map map = kafkaConsumer.listTopics();
//        System.out.println();
//
//    }



}

